package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.ods.*;
import cn.gwm.flink.streaming.ods.model.ModelCanEnum;
import cn.gwm.flink.streaming.ods.model.ModelGpsEnum;
import cn.gwm.utils.ConfigLoader;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * evOds phevOds 落库有效字段
 * @ClassName OdsThTask
 * @Author xzh
 * @Date 2022/12/15 15:46
 **/
public class OdsOrcTxtTask {

    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 3.1）设置每隔30s一个周期开启checkpoint（checkpoint开启周期，不能太长也不能太短：1-5分钟）
        env.enableCheckpointing(300 * 1000L);
        // 启动类型
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        // 3.2）设置检查点的mode，exactly-once，保证数据的一次性语义
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 3.3）设置两次checkpoint之间的时间间隔，避免两次间隔时间太近导致频繁的checkpoint
        env.getCheckpointConfig().setCheckpointInterval(300*1000L);
        // 3.4）设置checkpoint的超时时间（一般设置一个checkpoint周期的二分之一）
        env.getCheckpointConfig().setCheckpointTimeout(600 * 1000L);
        // 3.5）设置checkpoint的最大尝试次数，同一个时间内有几个checkpoint在运行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L);
        // 3.6）设置job作业取消的时候，是否保留checkpoint的计算结果
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 3.7）设置job作业运行过程中，如果checkpoint失败，是否停止作业
        env.getCheckpointConfig().setFailOnCheckpointingErrors(true);
        // 3.8）设置checkpoint的存储后端，使用rocksdb作为状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        // 设置可容忍的检查点的失败数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        //启用不对齐检查点保存方式
        env.getCheckpointConfig().enableUnalignedCheckpoints();
        env.getCheckpointConfig().setCheckpointStorage(ConfigLoader.get("hdfsUri") + "/flink/checkpoints/" + "OdsOrcTxtTask");
        env.setRestartStrategy(RestartStrategies.failureRateRestart(10, Time.minutes(15),Time.seconds(10)));
        Map<String, OutputTag<String>> canTagMap = new ConcurrentHashMap<>();
        for (ModelCanEnum canEnum : ModelCanEnum.values()) {
            canTagMap.put(canEnum.getModelName(), new OutputTag<String>(canEnum.getValue()) {
            });
        }
        Map<String, OutputTag<String>> gpsTagMap = new ConcurrentHashMap<>();
        for (ModelGpsEnum gpsEnum : ModelGpsEnum.values()) {
            gpsTagMap.put(gpsEnum.getModelName(), new OutputTag<String>(gpsEnum.getValue()) {
            });
        }
        String suffix = System.getenv().get(DefaultConstant.BOOT_PROFILES_ACTIVE);
        if (suffix == null) {
            suffix = "dev";
        }
        SingleOutputStreamOperator<String> gpsProcess = OdsPreTask.defindGpsStream(env, gpsTagMap,
                ConfigLoader.get("dataHub.gps.subId.toHive")).name("gpsStream");
        SingleOutputStreamOperator<String> canProcess = OdsPreTask.defindStream(env, canTagMap,
                ConfigLoader.get("dataHub.can.subId.toHive")).name("canStream");
        OdsHandle.handleEs13(canProcess,gpsProcess,canTagMap,gpsTagMap,suffix);
        OdsHandle.handleV71(canProcess,gpsProcess,canTagMap,gpsTagMap,suffix);
        if("dev".equals(suffix)){
//            OdsHandle.handleP03(canProcess,gpsProcess,canTagMap,gpsTagMap,suffix);
            OdsHandle.handleB07(canProcess,gpsProcess,canTagMap,gpsTagMap);
            OdsHandle.handleC01(canProcess,gpsProcess,canTagMap,gpsTagMap);
            OdsHandle.handleB01(canProcess,gpsProcess,canTagMap,gpsTagMap);
            OdsHandle.handleEc24(canProcess,gpsProcess,canTagMap,gpsTagMap);
            OdsHandle.handleEs24(canProcess,gpsProcess,canTagMap,gpsTagMap);
        }
        env.execute("OdsOrcTxtTask");
    }



}
